package cm.kafka.send;

import cm.kafka.configuration.ExecutorConfig;
import cm.kafka.constants.LogCollectionConstants;
import cm.kafka.entity.AuditLog;
import cm.kafka.entity.EventLog;
import cm.kafka.util.LocalDateTimeUtils;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.function.BiConsumer;

@Service
@Slf4j
public class KafkaConsumerDemo {
    @Autowired
    ExecutorConfig executorConfig;

    @Autowired
    KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    RestHighLevelClient restHighLevelClient;

    @KafkaListener(topics = "tsa-syslog_172.16.10.111")
    public void consumerTopic(List<ConsumerRecord> records, Acknowledgment ack) {
        ack.acknowledge();
        Thread thread = new Thread(() -> {
            sendMessage(records);
        });
        executorConfig.getExecutor().execute(thread);//丢进线程池管理
    }

    public void sendMessage(List<ConsumerRecord> records) {
        log.info("\n" + "-------开始数据结束,处理的条数：" + records.size() + "-------" + "\n");
        Long startTime = System.currentTimeMillis();
        List<AuditLog> auditLogList;
        //数据转化解析
        auditLogList = this.dohandle(records);
        doSend(auditLogList);
        log.info("----当前线程：" + Thread.currentThread().getName() + "----tsa-syslog分区:" + records.get(0).partition() + "-------总共耗时:" + (System.currentTimeMillis() - startTime) + "ms-----" + ",处理成功：" + auditLogList.size() + "\n");
        log.info("\n" + "-------处理数据结束-------" + "\n");
    }


    private List<AuditLog> dohandle(List<ConsumerRecord> records) {
        List<AuditLog> logList = new ArrayList<>();
        List<EventLog> eventLogList = new ArrayList<>();
        EventLog eventLog;
        for (ConsumerRecord record : records) {
            //message转化为EventLog对象
            eventLog = setNecessEventLogData(record);
            eventLogList.add(eventLog);
        }

        AuditLog auditLog;
        if (!CollectionUtils.isEmpty(eventLogList)) {
            //将分组的ip数据进行添加入库
            for (EventLog eve : eventLogList) {
                auditLog = new AuditLog(eve);
                logList.add(auditLog);
            }
        }

        return logList;
    }

    //message转化为EventLog对象
    private EventLog setNecessEventLogData(ConsumerRecord record) {
        JSONObject jsonObject = JSONObject.parseObject(record.value().toString());
        EventLog eventLog = new EventLog();
        eventLog.setEventAcceptTime(LocalDateTimeUtils.convertLDTToDate(LocalDateTime.now())); // 接收时间
        if (jsonObject.get("fields") != null) {
            String str = jsonObject.getJSONObject("fields").getString("host_ip");
            eventLog.setHost(str);
            eventLog.setDeviceAddress(str);
        }
        if (jsonObject.containsKey("collectionType")) {
            eventLog.setCollectionType(jsonObject.get("collectionType").toString());
        } else if (jsonObject.containsKey("type")) {
            eventLog.setCollectionType(jsonObject.get("type").toString());
        }

        eventLog.setEventMessage(jsonObject.get("eventMessage").toString());
        return eventLog;
    }

    private void doSend(List<AuditLog> auditLogList) {
        BulkRequest bulkRequest = new BulkRequest();
        LocalDate today = LocalDate.now();
        //如果索引是日志审计的，则索引名称追加日期 防止数据都存在一个索引中
        String indexName = LogCollectionConstants.LOG_INDEX_NAME_ALIAS + "_" + today.format(LogCollectionConstants.DATE_FORMATTER);
        for (AuditLog message : auditLogList) {
            if (message != null) {
                IndexRequest indexRequest = new IndexRequest(indexName).opType(DocWriteRequest.OpType.INDEX);
                indexRequest.source(JSONObject.toJSONString(message), XContentType.JSON);
                bulkRequest.add(indexRequest);
            }
        }

            BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer =
                    (request, bulkListener) -> restHighLevelClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);
//            if (bulkConsumer.hasFailures()) {
//                log.error("es插入存在错误");
//                log.error(bulk.buildFailureMessage());
//            }else{
//                log.info("ES插入总耗时："+ bulk.getTook().getSeconds() + "秒"+",成功转发条数:"+auditLogList.size());
//            }



    }


}
